package com.cnebula.dataprocess.sqoop.job;

import java.util.Map;

import org.apache.sqoop.model.MFromConfig;
import org.apache.sqoop.model.MJob;

import com.cnebula.dataprocess.sqoop.job.adapter.SqoopJobConfigAdapter;
import com.cnebula.dataprocess.sqoop.job.model.FromSqoopJobConstants;

public class JDBCJobConfigAdapter extends SqoopJobConfigAdapter {

	private static JDBCJobConfigAdapter instance = null;

	private static final String SQL_CONDITIONS_TOKEN = "${CONDITIONS}";

	static {
		instance = new JDBCJobConfigAdapter();
	}

	public static JDBCJobConfigAdapter getInstance() {
		return instance;
	}

	private JDBCJobConfigAdapter() {
	}

	public void setMFromConfig(MJob job, Map<String,String> jobConfig) {
		MFromConfig fromJobConfig = job.getFromJobConfig();
		String sql = this.formatSQL(jobConfig.get(FromSqoopJobConstants.JDBC_SQL));
		String partitionColumn = jobConfig.get(FromSqoopJobConstants.JDBC_PARTITIONCOLUMN);
		
		fromJobConfig.getStringInput("fromJobConfig.sql").setValue(sql);
		fromJobConfig.getStringInput("fromJobConfig.partitionColumn").setValue(partitionColumn);
	}

	@Override
	public void setMToConfig(MJob job, Map<String,String> jobConfig) {

	}

	private String formatSQL(String sql) {
		String rst = sql;

		String str = sql.toUpperCase();

		int pos = str.indexOf("WHERE");
		if (pos > 0) {
			
			String beginStr = rst.substring(0, pos);
			String endStr = rst.substring(pos + 5);
			rst = beginStr + " WHERE " + SQL_CONDITIONS_TOKEN + " AND " + endStr;
			
		} else if (str.indexOf("ORDER BY") > 0) {
			
			pos = str.indexOf("ORDER BY");
			String beginStr = rst.substring(0, pos);
			String endStr = rst.substring(pos);
			rst = beginStr + " WHERE " + SQL_CONDITIONS_TOKEN + " " + endStr;
		}else{
			rst = sql + " WHERE " + SQL_CONDITIONS_TOKEN;
		}

		return rst;
	}

}
